{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exploring tf.transform # \n",
    "\n",
    "**Learning Objectives**\n",
    "1. Preprocess data and engineer new features using TfTransform. \n",
    "1. Create and deploy Apache Beam pipeline. \n",
    "1. Use processed data to train taxifare model locally then serve a prediction.\n",
    "\n",
    "## Introduction \n",
    "While Pandas is fine for experimenting, for operationalization of your workflow it is better to do preprocessing in Apache Beam. This will also help if you need to preprocess data in flight, since Apache Beam allows for streaming. In this lab we will pull data from BigQuery then use Apache Beam  TfTransform to process the data.  \n",
    "\n",
    "Only specific combinations of TensorFlow/Beam are supported by tf.transform so make sure to get a combo that works. In this lab we will be using: \n",
    "* TFT 0.24.0\n",
    "* TF 2.3.0 \n",
    "* Apache Beam [GCP] 2.24.0\n",
    "\n",
    "Each learning objective will correspond to a __#TODO__ in the [student lab notebook](../labs/5_tftransform_taxifare.ipynb) -- try to complete that notebook first before reviewing this solution notebook."

   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run the chown command to change the ownership\n",
    "!sudo chown -R jupyter:jupyter /home/jupyter/training-data-analyst"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: tensorflow==2.3.0 in /opt/conda/lib/python3.7/site-packages (2.3.0)\n",
      "Collecting tensorflow-transform==0.24.0\n",
      "Downloading tensorflow_transform-0.24.0-py3-none-any.whl (373 kB)\n",
      " |████████████████████████████████| 373 kB 4.4 MB/s eta 0:00:01\n",
      "Collecting apache-beam[gcp]==2.24.0\n",
      "  Downloading apache_beam-2.24.0-cp37-cp37m-manylinux2010_x86_64.whl (8.5 MB)\n",
      "     |████████████████████████████████| 8.5 MB 8.6 MB/s eta 0:00:01\n",
      "Collecting scipy==1.4.1\n",
      "  Downloading scipy-1.4.1-cp37-cp37m-manylinux1_x86_64.whl (26.1 MB)\n",
      "     |████████████████████████████████| 26.1 MB 65.8 MB/s eta 0:00:01\n",
      "...\n",
      "...\n",
      "...\n",
      "Successfully built oauth2client\n",
      "Installing collected packages: httplib2, oauth2client, mock, pyarrow, cachetools, apache-beam, tensorflow-metadata, tfx-bsl, tensorflow-transform, scipy\n",
      "  Attempting uninstall: httplib2\n",
          "Found existing installation: httplib2 0.18.1\n",
          "Uninstalling httplib2-0.18.1:\n",
          "  Successfully uninstalled httplib2-0.18.1\n",
        "Attempting uninstall: oauth2client\n",
          "Found existing installation: oauth2client 4.1.3\n",
          "Uninstalling oauth2client-4.1.3:\n",
          "  Successfully uninstalled oauth2client-4.1.3\n",
        "Attempting uninstall: mock\n",
        "  Found existing installation: mock 4.0.2\n",
        "  Uninstalling mock-4.0.2:\n",
        "    Successfully uninstalled mock-4.0.2\n",
        "Attempting uninstall: pyarrow\n",
        "  Found existing installation: pyarrow 1.0.1\n",
        "  Uninstalling pyarrow-1.0.1:\n",
        "    Successfully uninstalled pyarrow-1.0.1\n",
        "Attempting uninstall: cachetools\n",
        "  Found existing installation: cachetools 4.1.1\n",
        "  Uninstalling cachetools-4.1.1:\n",
        "    Successfully uninstalled cachetools-4.1.1\n",
        "Attempting uninstall: apache-beam\n",
          "Found existing installation: apache-beam 2.23.0\n",
          "Uninstalling apache-beam-2.23.0:\n",
            "Successfully uninstalled apache-beam-2.23.0\n",
        "Attempting uninstall: tensorflow-metadata\n",
          "Found existing installation: tensorflow-metadata 0.23.0\n",
          "Uninstalling tensorflow-metadata-0.23.0:\n",
            "Successfully uninstalled tensorflow-metadata-0.23.0\n",
        "Attempting uninstall: tfx-bsl\n",
          "Found existing installation: tfx-bsl 0.23.0\n",
          "Uninstalling tfx-bsl-0.23.0:\n",
            "Successfully uninstalled tfx-bsl-0.23.0\n",
        "Attempting uninstall: tensorflow-transform\n",
          "Found existing installation: tensorflow-transform 0.23.0\n",
          "Uninstalling tensorflow-transform-0.23.0:\n",
            "Successfully uninstalled tensorflow-transform-0.23.0\n",
        "Attempting uninstall: scipy\n",
          "Found existing installation: scipy 1.5.2\n",
          "Uninstalling scipy-1.5.2:\n",
            "Successfully uninstalled scipy-1.5.2\n",
      "ERROR: After October 2020 you may experience errors when installing or updating packages. This is because pip will change the way that it resolves dependency conflicts.\n",
      "We recommend you use --use-feature=2020-resolver to test your packages with the new resolver before it becomes the default.\n",
      "witwidget 1.7.0 requires oauth2client>=4.1.3, but you'll have oauth2client 3.0.0 which is incompatible.\n",
      "tfx 0.23.0 requires attrs<20,>=19.3.0, but you'll have attrs 20.2.0 which is incompatible.\n",
      "tfx 0.23.0 requires google-resumable-media<0.7.0,>=0.6.0, but you'll have google-resumable-media 1.0.0 which is incompatible.\n",
      "tfx 0.23.0 requires tensorflow-transform<0.24,>=0.23, but you'll have tensorflow-transform 0.24.0 which is incompatible.\n",
      "tfx 0.23.0 requires tfx-bsl<0.24,>=0.23, but you'll have tfx-bsl 0.24.0 which is incompatible.\n",
      "tfx-bsl 0.24.0 requires absl-py<0.11,>=0.9, but you'll have absl-py 0.8.1 which is incompatible.\n",
      "tensorflow-transform 0.24.0 requires absl-py<0.11,>=0.9, but you'll have absl-py 0.8.1 which is incompatible.\n",
      "tensorflow-model-analysis 0.23.0 requires tensorflow-metadata<0.24,>=0.23, but you'll have tensorflow-metadata 0.24.0 which is incompatible.\n",
      "tensorflow-model-analysis 0.23.0 requires tfx-bsl<0.24,>=0.23, but you'll have tfx-bsl 0.24.0 which is incompatible.\n",
      "tensorflow-metadata 0.24.0 requires absl-py<0.11,>=0.9, but you'll have absl-py 0.8.1 which is incompatible.\n",
      "tensorflow-data-validation 0.23.0 requires joblib<0.15,>=0.12, but you'll have joblib 0.16.0 which is incompatible.\n",
      "tensorflow-data-validation 0.23.0 requires tensorflow-metadata<0.24,>=0.23, but you'll have tensorflow-metadata 0.24.0 which is incompatible.\n",
      "tensorflow-data-validation 0.23.0 requires tensorflow-transform<0.24,>=0.23, but you'll have tensorflow-transform 0.24.0 which is incompatible.\n",
      "tensorflow-data-validation 0.23.0 requires tfx-bsl<0.24,>=0.23, but you'll have tfx-bsl 0.24.0 which is incompatible.\n",
      "cloud-tpu-client 0.10 requires google-api-python-client==1.8.0, but you'll have google-api-python-client 1.11.0 which is incompatible.\n",
      "Successfully installed apache-beam-2.24.0 cachetools-3.1.1 httplib2-0.17.4 mock-2.0.0 oauth2client-3.0.0 pyarrow-0.17.1 scipy-1.4.1 tensorflow-metadata-0.24.0 tensorflow-transform-0.24.0 tfx-bsl-0.24.0\n"
     ]
    }
   ],
   "source": [
    "# Install the necessary dependencies\n",
    "!pip install tensorflow==2.3.0 tensorflow-transform==0.24.0 apache-beam[gcp]==2.24.0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**NOTE**: You may ignore specific incompatibility errors and warnings. These components and issues do not impact your ability to complete the lab.\n",
    "Download .whl file for tensorflow-transform. We will pass this file to Beam Pipeline Options so it is installed on the DataFlow workers "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install --user google-cloud-bigquery==1.25.0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting tensorflow-transform==0.24.0\n",
      "Using cached tensorflow_transform-0.24.0-py3-none-any.whl (373 kB)\n",
      "Saved ./tensorflow_transform-0.24.0-py3-none-any.whl\n",
      "Successfully downloaded tensorflow-transform\n"
     ]
    }
   ],
   "source": [
    "!pip download tensorflow-transform==0.24.0 --no-deps"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Restart the kernel</b> (click on the reload button above)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "apache-beam==2.24.0\n",
      "tensorflow @ file:///opt/conda/conda-bld/dlenv-tf-2-3-cpu_1599847782078/work/tensorflow-2.3.0-cp37-cp37m-linux_x86_64.whl\n",
      "tensorflow-data-validation==0.23.\n",
      "tensorflow-datasets==3.2.1\n",
      "tensorflow-enterprise-addons @ file:///opt/conda/conda-bld/dlenv-tf-2-3-cpu_1599847782078/work/tensorflow_enterprise_addons-0.0.0-py3-none-any.whl\n",
      "tensorflow-estimator==2.3.0\n",
      "tensorflow-hub==0.9.0\n",
      "tensorflow-io==0.15.0\n",
      "tensorflow-metadata==0.24.0\n",
      "tensorflow-model-analysis==0.23.0\n",
      "tensorflow-probability==0.11.0\n",
      "tensorflow-serving-api==2.3.0\n",
      "tensorflow-transform==0.24.0\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "# Output installed packages in requirements format.\n",
    "pip freeze | grep -e 'flow\\|beam'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2.3.0\n"
     ]
    }
   ],
   "source": [
    "# Import data processing libraries\n",
    "import tensorflow as tf\n",
    "import tensorflow_transform as tft\n",
    "# Python shutil module enables us to operate with file objects easily and without diving into file objects a lot.\n",
    "import shutil\n",
    "# Show the currently installed version of TensorFlow\n",
    "print(tf.__version__)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# change these to try this notebook out\n",
    "BUCKET = 'cloud-example-labs'\n",
    "PROJECT = 'project-id'\n",
    "REGION = 'us-central1'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The OS module in python provides functions for interacting with the operating system.\n",
    "import os\n",
    "os.environ['BUCKET'] = BUCKET\n",
    "os.environ['PROJECT'] = PROJECT\n",
    "os.environ['REGION'] = REGION"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Updated property [core/project].\n",
      "Updated property [compute/region].\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "# gcloud config set - set a Cloud SDK property\n",
    "gcloud config set project $PROJECT\n",
    "gcloud config set compute/region $REGION"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Creating gs://qwiklabs-gcp-xxxxxxxx/...\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "# Create bucket\n",
    "if ! gcloud storage ls | grep -q gs://${BUCKET}/; then\n",    "  gcloud storage buckets create --location=${REGION} gs://${BUCKET}\n",    "fi"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Input source: BigQuery\n",
    "\n",
    "Get data from BigQuery but defer the majority of filtering etc. to Beam.\n",
    "Note that the dayofweek column is now strings."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import Google BigQuery API client library\n",
    "from google.cloud import bigquery\n",
    "\n",
    "\n",
    "def create_query(phase, EVERY_N):\n",
    "    \"\"\"Creates a query with the proper splits.\n",
    "\n",
    "    Args:\n",
    "        phase: int, 1=train, 2=valid.\n",
    "        EVERY_N: int, take an example EVERY_N rows.\n",
    "\n",
    "    Returns:\n",
    "        Query string with the proper splits.\n",
    "    \"\"\"\n",
    "    base_query = \"\"\"\n",
    "    WITH daynames AS\n",
    "    (SELECT ['Sun', 'Mon', 'Tues', 'Wed', 'Thurs', 'Fri', 'Sat'] AS daysofweek)\n",
    "    SELECT\n",
    "    (tolls_amount + fare_amount) AS fare_amount,\n",
    "    daysofweek[ORDINAL(EXTRACT(DAYOFWEEK FROM pickup_datetime))] AS dayofweek,\n",
    "    EXTRACT(HOUR FROM pickup_datetime) AS hourofday,\n",
    "    pickup_longitude AS pickuplon,\n",
    "    pickup_latitude AS pickuplat,\n",
    "    dropoff_longitude AS dropofflon,\n",
    "    dropoff_latitude AS dropofflat,\n",
    "    passenger_count AS passengers,\n",
    "    'notneeded' AS key\n",
    "    FROM\n",
    "    `nyc-tlc.yellow.trips`, daynames\n",
    "    WHERE\n",
    "    trip_distance > 0 AND fare_amount > 0\n",
    "    \"\"\"\n",
    "    if EVERY_N is None:\n",
    "        if phase < 2:\n",
    "            # training\n",
    "            query = \"\"\"{0} AND ABS(MOD(FARM_FINGERPRINT(CAST\n",
    "            (pickup_datetime AS STRING), 4)) < 2\"\"\".format(base_query)\n",
    "        else:\n",
    "            query = \"\"\"{0} AND ABS(MOD(FARM_FINGERPRINT(CAST(\n",
    "            pickup_datetime AS STRING), 4)) = {1}\"\"\".format(base_query, phase)\n",
    "    else:\n",
    "        query = \"\"\"{0} AND ABS(MOD(FARM_FINGERPRINT(CAST(\n",
    "        pickup_datetime AS STRING)), {1})) = {2}\"\"\".format(\n",
    "            base_query, EVERY_N, phase)\n",
    "\n",
    "    return query\n",
    "\n",
    "query = create_query(2, 100000)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's pull this query down into a Pandas DataFrame and take a look at some of the statistics."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>fare_amount</th>\n",
       "      <th>dayofweek</th>\n",
       "      <th>hourofday</th>\n",
       "      <th>pickuplon</th>\n",
       "      <th>pickuplat</th>\n",
       "      <th>dropofflon</th>\n",
       "      <th>dropofflat</th>\n",
       "      <th>passengers</th>\n",
       "      <th>key</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>4.5</td>\n",
       "      <td>Sat</td>\n",
       "      <td>0</td>\n",
       "      <td>-74.000292</td>\n",
       "      <td>40.728722</td>\n",
       "      <td>-73.995235</td>\n",
       "      <td>40.724961</td>\n",
       "      <td>1</td>\n",
       "      <td>notneeded</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>6.9</td>\n",
       "      <td>Sun</td>\n",
       "      <td>0</td>\n",
       "      <td>-73.986003</td>\n",
       "      <td>40.722688</td>\n",
       "      <td>-74.004549</td>\n",
       "      <td>40.718822</td>\n",
       "      <td>1</td>\n",
       "      <td>notneeded</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>16.5</td>\n",
       "      <td>Sun</td>\n",
       "      <td>0</td>\n",
       "      <td>-74.002155</td>\n",
       "      <td>40.740375</td>\n",
       "      <td>-73.967537</td>\n",
       "      <td>40.792845</td>\n",
       "      <td>5</td>\n",
       "      <td>notneeded</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>143.0</td>\n",
       "      <td>Sun</td>\n",
       "      <td>0</td>\n",
       "      <td>-73.990255</td>\n",
       "      <td>40.740407</td>\n",
       "      <td>-74.350245</td>\n",
       "      <td>40.663847</td>\n",
       "      <td>1</td>\n",
       "      <td>notneeded</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>19.0</td>\n",
       "      <td>Sun</td>\n",
       "      <td>0</td>\n",
       "      <td>-73.977255</td>\n",
       "      <td>40.754930</td>\n",
       "      <td>-73.917570</td>\n",
       "      <td>40.767272</td>\n",
       "      <td>1</td>\n",
       "      <td>notneeded</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   fare_amount dayofweek  hourofday  pickuplon  pickuplat  dropofflon  \\\n",
       "0          4.5       Sat          0 -74.000292  40.728722  -73.995235   \n",
       "1          6.9       Sun          0 -73.986003  40.722688  -74.004549   \n",
       "2         16.5       Sun          0 -74.002155  40.740375  -73.967537   \n",
       "3        143.0       Sun          0 -73.990255  40.740407  -74.350245   \n",
       "4         19.0       Sun          0 -73.977255  40.754930  -73.917570   \n",
       "\n",
       "   dropofflat  passengers        key  \n",
       "0   40.724961           1  notneeded  \n",
       "1   40.718822           1  notneeded  \n",
       "2   40.792845           5  notneeded  \n",
       "3   40.663847           1  notneeded  \n",
       "4   40.767272           1  notneeded  "
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>fare_amount</th>\n",
       "      <th>hourofday</th>\n",
       "      <th>pickuplon</th>\n",
       "      <th>pickuplat</th>\n",
       "      <th>dropofflon</th>\n",
       "      <th>dropofflat</th>\n",
       "      <th>passengers</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>count</th>\n",
       "      <td>11181.000000</td>\n",
       "      <td>11181.000000</td>\n",
       "      <td>11181.000000</td>\n",
       "      <td>11181.000000</td>\n",
       "      <td>11181.000000</td>\n",
       "      <td>11181.000000</td>\n",
       "      <td>11181.000000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>mean</th>\n",
       "      <td>11.242599</td>\n",
       "      <td>13.244075</td>\n",
       "      <td>-72.576852</td>\n",
       "      <td>39.973146</td>\n",
       "      <td>-72.748974</td>\n",
       "      <td>40.006091</td>\n",
       "      <td>1.722118</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>std</th>\n",
       "      <td>9.447462</td>\n",
       "      <td>6.548354</td>\n",
       "      <td>10.133452</td>\n",
       "      <td>5.777329</td>\n",
       "      <td>12.981577</td>\n",
       "      <td>5.664887</td>\n",
       "      <td>1.351062</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>min</th>\n",
       "      <td>2.500000</td>\n",
       "      <td>0.000000</td>\n",
       "      <td>-78.133333</td>\n",
       "      <td>-73.991278</td>\n",
       "      <td>-751.400000</td>\n",
       "      <td>-73.977970</td>\n",
       "      <td>0.000000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>25%</th>\n",
       "      <td>6.000000</td>\n",
       "      <td>9.000000</td>\n",
       "      <td>-73.991849</td>\n",
       "      <td>40.734954</td>\n",
       "      <td>-73.991236</td>\n",
       "      <td>40.734008</td>\n",
       "      <td>1.000000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>50%</th>\n",
       "      <td>8.500000</td>\n",
       "      <td>14.000000</td>\n",
       "      <td>-73.981824</td>\n",
       "      <td>40.752640</td>\n",
       "      <td>-73.980164</td>\n",
       "      <td>40.753427</td>\n",
       "      <td>1.000000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>75%</th>\n",
       "      <td>12.500000</td>\n",
       "      <td>19.000000</td>\n",
       "      <td>-73.967418</td>\n",
       "      <td>40.766700</td>\n",
       "      <td>-73.964153</td>\n",
       "      <td>40.767832</td>\n",
       "      <td>2.000000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>max</th>\n",
       "      <td>143.000000</td>\n",
       "      <td>23.000000</td>\n",
       "      <td>40.806487</td>\n",
       "      <td>41.366138</td>\n",
       "      <td>40.785400</td>\n",
       "      <td>41.366138</td>\n",
       "      <td>6.000000</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        fare_amount     hourofday     pickuplon     pickuplat    dropofflon  \\\n",
       "count  11181.000000  11181.000000  11181.000000  11181.000000  11181.000000   \n",
       "mean      11.242599     13.244075    -72.576852     39.973146    -72.748974   \n",
       "std        9.447462      6.548354     10.133452      5.777329     12.981577   \n",
       "min        2.500000      0.000000    -78.133333    -73.991278   -751.400000   \n",
       "25%        6.000000      9.000000    -73.991849     40.734954    -73.991236   \n",
       "50%        8.500000     14.000000    -73.981824     40.752640    -73.980164   \n",
       "75%       12.500000     19.000000    -73.967418     40.766700    -73.964153   \n",
       "max      143.000000     23.000000     40.806487     41.366138     40.785400   \n",
       "\n",
       "         dropofflat    passengers  \n",
       "count  11181.000000  11181.000000  \n",
       "mean      40.006091      1.722118  \n",
       "std        5.664887      1.351062  \n",
       "min      -73.977970      0.000000  \n",
       "25%       40.734008      1.000000  \n",
       "50%       40.753427      1.000000  \n",
       "75%       40.767832      2.000000  \n",
       "max       41.366138      6.000000  "
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_valid = bigquery.Client().query(query).to_dataframe()\n",
    "# `head()` function is used to get the first n rows of dataframe\n",
    "display(df_valid.head())\n",
    "# `describe()` is use to get the statistical summary of the DataFrame\n",
    "df_valid.describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create ML dataset using tf.transform and Dataflow\n",
    "\n",
    "Let's use Cloud Dataflow to read in the BigQuery data and write it out as TFRecord files. Along the way, let's use tf.transform to do scaling and transforming. Using tf.transform allows us to save the metadata to ensure that the appropriate transformations get carried out during prediction as well."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`transformed_data` is type `pcollection`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Launching Dataflow job preprocess-taxi-features-200924-103136 ... hang on\n",
      "WARNING:tensorflow:Tensorflow version (2.3.0) found. Note that Tensorflow Transform support for TF 2.0 is currently in beta, and features such as tf.function may not work as intended.\n",
      "WARNING:tensorflow:Tensorflow version (2.3.0) found. Note that Tensorflow Transform support for TF 2.0 is currently in beta, and features such as tf.function may not work as intended.\n",
      "WARNING:tensorflow:Tensorflow version (2.3.0) found. Note that Tensorflow Transform support for TF 2.0 is currently in beta, and features such as tf.function may not work as intended.\n",
      "WARNING:tensorflow:Tensorflow version (2.3.0) found. Note that Tensorflow Transform support for TF 2.0 is currently in beta, and features such as tf.function may not work as intended.\n",
      "WARNING:tensorflow:You are passing instance dicts and DatasetMetadata to TFT which will not provide optimal performance. Consider following the TFT guide to upgrade to the TFXIO format (Apache Arrow RecordBatch).\n",
      "WARNING:tensorflow:You are passing instance dicts and DatasetMetadata to TFT which will not provide optimal performance. Consider following the TFT guide to upgrade to the TFXIO format (Apache Arrow RecordBatch).\n",
      "{'dayofweek': <tf.Tensor 'inputs/inputs/dayofweek/Placeholder_copy:0' shape=(None,) dtype=string>, 'dropofflat': <tf.Tensor 'inputs/inputs/dropofflat/Placeholder_copy:0' shape=(None,) dtype=float32>, 'dropofflon': <tf.Tensor 'inputs/inputs/dropofflon/Placeholder_copy:0' shape=(None,) dtype=float32>, 'fare_amount': <tf.Tensor 'inputs/inputs/F_fare_amount/Placeholder_copy:0' shape=(None,) dtype=float32>, 'hourofday': <tf.Tensor 'inputs/inputs/hourofday/Placeholder_copy:0' shape=(None,)\n", "dtype=int64>, 'key': <tf.Tensor 'inputs/inputs/key/Placeholder_copy:0' shape=(None,) dtype=string>, 'passengers': <tf.Tensor 'inputs/inputs/passengers/Placeholder_copy:0' shape=(None,) dtype=int64>, 'pickuplat': <tf.Tensor 'inputs/inputs/pickuplat/Placeholder_copy:0' shape=(None,) dtype=float32>, 'pickuplon': <tf.Tensor 'inputs/inputs/pickuplon/Placeholder_copy:0' shape=(None,) dtype=float32>}\n",
      "INFO:tensorflow:Assets added to graph.\n",
      "INFO:tensorflow:Assets added to graph.\n",
      "INFO:tensorflow:No assets to write.\n",
      "INFO:tensorflow:No assets to write.\n",
      "WARNING:tensorflow:Issue encountered when serializing tft_mapper_use.\n",
      "Type is unsupported, or the types of the items don't match field type in CollectionDef. Note this is a warning and probably safe to ignore.\n",
      "'Counter' object has no attribute 'name'\n",
      "WARNING:tensorflow:Issue encountered when serializing tft_mapper_use.\n",
      "Type is unsupported, or the types of the items don't match field type in CollectionDef. Note this is a warning and probably safe to ignore.\n",
      "'Counter' object has no attribute 'name'\n",
      "INFO:tensorflow:SavedModel written to: gs://qwiklabs-gcp-01-703c5c40c134/taxifare/preproc_tft/tmp/tftransform_tmp/fa360e660c0a4f85b4a07a9a6636e87f/saved_model.pb\n",
      "INFO:tensorflow:SavedModel written to: gs://qwiklabs-gcp-01-703c5c40c134/taxifare/preproc_tft/tmp/tftransform_tmp/fa360e660c0a4f85b4a07a9a6636e87f/saved_model.pb\n",
      "INFO:tensorflow:Assets added to graph.\n",
      "INFO:tensorflow:Assets added to graph.\n",
      "INFO:tensorflow:No assets to write.\n",
      "INFO:tensorflow:No assets to write.\n",
      "WARNING:tensorflow:Issue encountered when serializing tft_mapper_use.\n",
      "Type is unsupported, or the types of the items don't match field type in CollectionDef. Note this is a warning and probably safe to ignore.\n",
      "'Counter' object has no attribute 'name'\n",
      "WARNING:tensorflow:Issue encountered when serializing tft_mapper_use.\n",
      "Type is unsupported, or the types of the items don't match field type in CollectionDef. Note this is a warning and probably safe to ignore.\n",
      "'Counter' object has no attribute 'name'\n",
      "INFO:tensorflow:SavedModel written to: gs://qwiklabs-gcp-01-703c5c40c134/taxifare/preproc_tft/tmp/tftransform_tmp/84246834829e4666b0ab6cb50768ef17/saved_model.pb\n",
      "INFO:tensorflow:SavedModel written to: gs://qwiklabs-gcp-01-703c5c40c134/taxifare/preproc_tft/tmp/tftransform_tmp/84246834829e4666b0ab6cb50768ef17/saved_model.pb\n",
      "WARNING:tensorflow:Tensorflow version (2.3.0) found. Note that Tensorflow Transform support for TF 2.0 is currently in beta, and features such as tf.function may not work as intended.\n",
      "WARNING:tensorflow:Tensorflow version (2.3.0) found. Note that Tensorflow Transform support for TF 2.0 is currently in beta, and features such as tf.function may not work as intended.\n",
      "WARNING:tensorflow:You are passing instance dicts and DatasetMetadata to TFT which will not provide optimal performance. Consider following the TFT guide to upgrade to the TFXIO format (Apache Arrow RecordBatch).\n",
      "WARNING:tensorflow:You are passing instance dicts and DatasetMetadata to TFT which will not provide optimal performance. Consider following the TFT guide to upgrade to the TFXIO format (Apache Arrow RecordBatch).\n",
      "WARNING:tensorflow:Tensorflow version (2.3.0) found. Note that Tensorflow Transform support for TF 2.0 is currently in beta, and features such as tf.function may not work as intended.\n",
      "WARNING:tensorflow:Tensorflow version (2.3.0) found. Note that Tensorflow Transform support for TF 2.0 is currently in beta, and features such as tf.function may not work as intended.\n",
      "WARNING:tensorflow:You are passing instance dicts and DatasetMetadata to TFT which will not provide optimal performance. Consider following the TFT guide to upgrade to the TFXIO format (Apache Arrow RecordBatch).\n",
      "WARNING:tensorflow:You are passing instance dicts and DatasetMetadata to TFT which will not provide optimal performance. Consider following the TFT guide to upgrade to the TFXIO format (Apache Arrow RecordBatch).\n",
      "WARNING:apache_beam.runners.portability.stager:The .whl package tensorflow_transform-0.24.0-py3-none-any.whl is provided in --extra_package. This functionality is not officially supported. Since wheel packages are binary distributions, this package must be binary-compatible with the worker environment (e.g. Python 2.7 running on an x64 Linux host).\n",
      "WARNING:root:Make sure that locally built Python SDK docker image has Python 3.7 interpreter.\n",
      "WARNING:apache_beam.options.pipeline_options:Discarding invalid overrides: {'teardown_policy': 'TEARDOWN_ALWAYS', 'no_save_main_session': True}\n",
      "WARNING:apache_beam.options.pipeline_options:Discarding invalid overrides: {'teardown_policy': 'TEARDOWN_ALWAYS', 'no_save_main_session': True}\n"
     ]
    }
   ],
   "source": [
    "# Import a module named `datetime` to work with dates as date objects.\n",
    "import datetime\n",
    "# Import data processing libraries and modules\n",
    "import tensorflow as tf\n",
    "import apache_beam as beam\n",
    "import tensorflow_transform as tft\n",
    "import tensorflow_metadata as tfmd\n",
    "from tensorflow_transform.beam import impl as beam_impl\n",
    "\n",
    "\n",
    "def is_valid(inputs):\n",
    "    \"\"\"Check to make sure the inputs are valid.\n",
    "\n",
    "    Args:\n",
    "        inputs: dict, dictionary of TableRow data from BigQuery.\n",
    "\n",
    "    Returns:\n",
    "        True if the inputs are valid and False if they are not.\n",
    "    \"\"\"\n",
    "    try:\n",
    "        pickup_longitude = inputs['pickuplon']\n",
    "        dropoff_longitude = inputs['dropofflon']\n",
    "        pickup_latitude = inputs['pickuplat']\n",
    "        dropoff_latitude = inputs['dropofflat']\n",
    "        hourofday = inputs['hourofday']\n",
    "        dayofweek = inputs['dayofweek']\n",
    "        passenger_count = inputs['passengers']\n",
    "        fare_amount = inputs['fare_amount']\n",
    "        return fare_amount >= 2.5 and pickup_longitude > -78 \\\n",
    "            and pickup_longitude < -70 and dropoff_longitude > -78 \\\n",
    "            and dropoff_longitude < -70 and pickup_latitude > 37 \\\n",
    "            and pickup_latitude < 45 and dropoff_latitude > 37 \\\n",
    "            and dropoff_latitude < 45 and passenger_count > 0\n",
    "    except:\n",
    "        return False\n",
    "\n",
    "\n",
    "def preprocess_tft(inputs):\n",
    "    \"\"\"Preprocess the features and add engineered features with tf transform.\n",
    "\n",
    "    Args:\n",
    "        dict, dictionary of TableRow data from BigQuery.\n",
    "\n",
    "    Returns:\n",
    "        Dictionary of preprocessed data after scaling and feature engineering.\n",
    "    \"\"\"\n",
    "    import datetime\n",
    "    print(inputs)\n",
    "    result = {}\n",
    "    result['fare_amount'] = tf.identity(inputs['fare_amount'])\n",
    "    # build a vocabulary\n",
    "    # TODO 1\n",
    "    result['dayofweek'] = tft.string_to_int(inputs['dayofweek'])\n",
    "    result['hourofday'] = tf.identity(inputs['hourofday'])  # pass through\n",
    "    # scaling numeric values\n",
    "    # TODO 2\n",
    "    result['pickuplon'] = (tft.scale_to_0_1(inputs['pickuplon']))\n",
    "    result['pickuplat'] = (tft.scale_to_0_1(inputs['pickuplat']))\n",
    "    result['dropofflon'] = (tft.scale_to_0_1(inputs['dropofflon']))\n",
    "    result['dropofflat'] = (tft.scale_to_0_1(inputs['dropofflat']))\n",
    "    result['passengers'] = tf.cast(inputs['passengers'], tf.float32)  # a cast\n",
    "    # arbitrary TF func\n",
    "    result['key'] = tf.as_string(tf.ones_like(inputs['passengers']))\n",
    "    # engineered features\n",
    "    latdiff = inputs['pickuplat'] - inputs['dropofflat']\n",
    "    londiff = inputs['pickuplon'] - inputs['dropofflon']\n",
    "    # Scale our engineered features latdiff and londiff between 0 and 1\n",
    "    # TODO 3\n",
    "    result['latdiff'] = tft.scale_to_0_1(latdiff)\n",
    "    result['londiff'] = tft.scale_to_0_1(londiff)\n",
    "    dist = tf.sqrt(latdiff * latdiff + londiff * londiff)\n",
    "    result['euclidean'] = tft.scale_to_0_1(dist)\n",
    "    return result\n",
    "\n",
    "\n",
    "def preprocess(in_test_mode):\n",
    "    \"\"\"Sets up preprocess pipeline.\n",
    "\n",
    "    Args:\n",
    "        in_test_mode: bool, False to launch DataFlow job, True to run locally.\n",
    "    \"\"\"\n",
    "    import os\n",
    "    import os.path\n",
    "    import tempfile\n",
    "    from apache_beam.io import tfrecordio\n",
    "    from tensorflow_transform.coders import example_proto_coder\n",
    "    from tensorflow_transform.tf_metadata import dataset_metadata\n",
    "    from tensorflow_transform.tf_metadata import dataset_schema\n",
    "    from tensorflow_transform.beam import tft_beam_io\n",
    "    from tensorflow_transform.beam.tft_beam_io import transform_fn_io\n",
    "\n",
    "    job_name = 'preprocess-taxi-features' + '-'\n",
    "    job_name += datetime.datetime.now().strftime('%y%m%d-%H%M%S')\n",
    "    if in_test_mode:\n",
    "        import shutil\n",
    "        print('Launching local job ... hang on')\n",
    "        OUTPUT_DIR = './preproc_tft'\n",
    "        shutil.rmtree(OUTPUT_DIR, ignore_errors=True)\n",
    "        EVERY_N = 100000\n",
    "    else:\n",
    "        print('Launching Dataflow job {} ... hang on'.format(job_name))\n",
    "        OUTPUT_DIR = 'gs://{0}/taxifare/preproc_tft/'.format(BUCKET)\n",
    "        import subprocess\n",
    "        subprocess.call('gcloud storage rm --recursive {}'.format(OUTPUT_DIR).split())\n",    "        EVERY_N = 10000\n",
    "\n",
    "    options = {\n",
    "        'staging_location': os.path.join(OUTPUT_DIR, 'tmp', 'staging'),\n",
    "        'temp_location': os.path.join(OUTPUT_DIR, 'tmp'),\n",
    "        'job_name': job_name,\n",
    "        'project': PROJECT,\n",
    "        'num_workers': 1,\n",
    "        'max_num_workers': 1,\n",
    "        'teardown_policy': 'TEARDOWN_ALWAYS',\n",
    "        'no_save_main_session': True,\n",
    "        'direct_num_workers': 1,\n",
    "        'extra_packages': ['tensorflow_transform-0.24.0-py3-none-any.whl']\n",
    "        }\n",
    "\n",
    "    opts = beam.pipeline.PipelineOptions(flags=[], **options)\n",
    "    if in_test_mode:\n",
    "        RUNNER = 'DirectRunner'\n",
    "    else:\n",
    "        RUNNER = 'DataflowRunner'\n",
    "\n",
    "    # Set up raw data metadata\n",
    "    raw_data_schema = {\n",
    "        colname: dataset_schema.ColumnSchema(\n",
    "            tf.string, [], dataset_schema.FixedColumnRepresentation())\n",
    "        for colname in 'dayofweek,key'.split(',')\n",
    "    }\n",
    "\n",
    "    raw_data_schema.update({\n",
    "        colname: dataset_schema.ColumnSchema(\n",
    "            tf.float32, [], dataset_schema.FixedColumnRepresentation())\n",
    "        for colname in\n",
    "        'fare_amount,pickuplon,pickuplat,dropofflon,dropofflat'.split(',')\n",
    "    })\n",
    "\n",
    "    raw_data_schema.update({\n",
    "        colname: dataset_schema.ColumnSchema(\n",
    "            tf.int64, [], dataset_schema.FixedColumnRepresentation())\n",
    "        for colname in 'hourofday,passengers'.split(',')\n",
    "    })\n",
    "\n",
    "    raw_data_metadata = dataset_metadata.DatasetMetadata(\n",
    "        dataset_schema.Schema(raw_data_schema))\n",
    "\n",
    "    # Run Beam\n",
    "    with beam.Pipeline(RUNNER, options=opts) as p:\n",
    "        with beam_impl.Context(temp_dir=os.path.join(OUTPUT_DIR, 'tmp')):\n",
    "            # Save the raw data metadata\n",
    "            (raw_data_metadata |\n",
    "                'WriteInputMetadata' >> tft_beam_io.WriteMetadata(\n",
    "                    os.path.join(\n",
    "                        OUTPUT_DIR, 'metadata/rawdata_metadata'), pipeline=p))\n",
    "\n",
    "            # Read training data from bigquery and filter rows\n",
    "            raw_data = (p | 'train_read' >> beam.io.Read(\n",
    "                    beam.io.BigQuerySource(\n",
    "                        query=create_query(1, EVERY_N),\n",
    "                        use_standard_sql=True)) |\n",
    "                        'train_filter' >> beam.Filter(is_valid))\n",
    "\n",
    "            raw_dataset = (raw_data, raw_data_metadata)\n",
    "\n",
    "            # Analyze and transform training data\n",
    "            # TODO 4\n",
    "            transformed_dataset, transform_fn = (\n",
    "                raw_dataset | beam_impl.AnalyzeAndTransformDataset(\n",
    "                    preprocess_tft))\n",
    "            transformed_data, transformed_metadata = transformed_dataset\n",
    "\n",
    "            # Save transformed train data to disk in efficient tfrecord format\n",
    "            transformed_data | 'WriteTrainData' >> tfrecordio.WriteToTFRecord(\n",
    "                os.path.join(OUTPUT_DIR, 'train'), file_name_suffix='.gz',\n",
    "                coder=example_proto_coder.ExampleProtoCoder(\n",
    "                    transformed_metadata.schema))\n",
    "\n",
    "            # Read eval data from bigquery and filter rows\n",
    "            # TODO 5\n",
    "            raw_test_data = (p | 'eval_read' >> beam.io.Read(\n",
    "                beam.io.BigQuerySource(\n",
    "                    query=create_query(2, EVERY_N),\n",
    "                    use_standard_sql=True)) | 'eval_filter' >> beam.Filter(\n",
    "                        is_valid))\n",
    "\n",
    "            raw_test_dataset = (raw_test_data, raw_data_metadata)\n",
    "\n",
    "            # Transform eval data\n",
    "            transformed_test_dataset = (\n",
    "                (raw_test_dataset, transform_fn) | beam_impl.TransformDataset()\n",
    "                )\n",
    "            transformed_test_data, _ = transformed_test_dataset\n",
    "\n",
    "            # Save transformed train data to disk in efficient tfrecord format\n",
    "            (transformed_test_data |\n",
    "                'WriteTestData' >> tfrecordio.WriteToTFRecord(\n",
    "                    os.path.join(OUTPUT_DIR, 'eval'), file_name_suffix='.gz',\n",
    "                    coder=example_proto_coder.ExampleProtoCoder(\n",
    "                        transformed_metadata.schema)))\n",
    "\n",
    "            # Save transformation function to disk for use at serving time\n",
    "            (transform_fn |\n",
    "                'WriteTransformFn' >> transform_fn_io.WriteTransformFn(\n",
    "                    os.path.join(OUTPUT_DIR, 'metadata')))\n",
    "\n",
    "# Change to True to run locally\n",
    "preprocess(in_test_mode=False)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This will take __10-15 minutes__. You cannot go on in this lab until your DataFlow job has successfully completed. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Note**: The above command may fail with an error **`Workflow failed. Causes: There was a problem refreshing your credentials`**. In that case, `re-run` the command again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "gs://cloud-example-labs/taxifare/preproc_tft/\n",
      "gs://cloud-example-labs/taxifare/preproc_tft/eval-00000-of-00001.gz\n",
      "gs://cloud-example-labs/taxifare/preproc_tft/train-00000-of-00004.gz\n",
      "gs://cloud-example-labs/taxifare/preproc_tft/train-00001-of-00004.gz\n",
      "gs://cloud-example-labs/taxifare/preproc_tft/train-00002-of-00004.gz\n",
      "gs://cloud-example-labs/taxifare/preproc_tft/train-00003-of-00004.gz\n",
      "gs://cloud-example-labs/taxifare/preproc_tft/metadata/\n",
      "gs://cloud-example-labs/taxifare/preproc_tft/tmp/\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "# ls preproc_tft\n",
    "# `ls` command show the full list or content of your directory\n",
    "gcloud storage ls gs://${BUCKET}/taxifare/preproc_tft/"   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Train off preprocessed data ##\n",
    "Now that we have our data ready and verified it is in the correct location we can train our taxifare model locally."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "rm: cannot remove './taxi_trained': No such file or directory\n",
      "INFO:tensorflow:Using default config.\n",
      "INFO:tensorflow:Using config: {'_session_config': allow_soft_placement: true\n",
      "graph_options {\n",
      "  rewrite_options {\n",
      "    meta_optimizer_iterations: ONE\n",
      "  }\n",
      "}\n",
      ", '_log_step_count_steps': 100, '_device_fn': None, '_service': None, '_model_dir': './taxi_trained', '_experimental_distribute': None, '_protocol': None, '_is_chief': True, '_task_id': 0, '_cluster_spec': <tensorflow.python.training.server_lib.ClusterSpec object at 0x7f15433f00f0>, '_task_type': 'worker', '_evaluation_master': '', '_master': '', '_keep_checkpoint_every_n_hours': 10000, '_tf_random_seed': None, '_keep_checkpoint_max': 5, '_save_checkpoints_secs': 600, '_save_checkpoints_steps': None, '_num_ps_replicas': 0, '_train_distribute': None, '_save_summary_steps': 100, '_eval_distribute': None, '_experimental_max_worker_delay_secs': None, '_num_worker_replicas': 1, '_global_id_in_cluster': 0, '_session_creation_timeout_secs': 7200}\n",
      "INFO:tensorflow:Not using Distribute Coordinator.\n",
      "INFO:tensorflow:Running training and evaluation locally (non-distributed).\n",
      "INFO:tensorflow:Start train and evaluate loop. The evaluate will happen after every checkpoint. Checkpoint frequency is determined based on RunConfig arguments: save_checkpoints_steps None or save_checkpoints_secs 600.\n",
      "WARNING:tensorflow:From /usr/local/lib/python3.5/dist-packages/tensorflow_core/python/ops/resource_variable_ops.py:1630: calling BaseResourceVariable.__init__ (from tensorflow.python.ops.resource_variable_ops) with constraint is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "If using Keras pass *_constraint arguments to layers.\n",
      "WARNING:tensorflow:From /usr/local/lib/python3.5/dist-packages/tensorflow_core/python/training/training_util.py:236: Variable.initialized_value (from tensorflow.python.ops.variables) is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "Use Variable.read_value. Variables in 2.X are initialized automatically both in eager and graph (inside tf.defun) contexts.\n",
      "INFO:tensorflow:Calling model_fn.\n",
      "WARNING:tensorflow:From /usr/local/lib/python3.5/dist-packages/tensorflow_estimator/python/estimator/head/regression_head.py:156: to_float (from tensorflow.python.ops.math_ops) is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "Use `tf.cast` instead.\n",
      "WARNING:tensorflow:From /usr/local/lib/python3.5/dist-packages/tensorflow_core/python/keras/optimizer_v2/adagrad.py:108: calling Constant.__init__ (from tensorflow.python.ops.init_ops) with dtype is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "Call initializer instance with the dtype argument instead of passing it to the constructor\n",
      "INFO:tensorflow:Done calling model_fn.\n",
      "INFO:tensorflow:Create CheckpointSaverHook.\n",
      "INFO:tensorflow:Graph was finalized.\n",
      "2019-12-17 21:48:33.822785: I tensorflow/core/platform/cpu_feature_guard.cc:145] This TensorFlow binary is optimized with Intel(R) MKL-DNN to use the following CPU instructions in performance critical operations:  AVX2 FMA\n",
      "To enable them in non-MKL-DNN operations, rebuild TensorFlow with the appropriate compiler flags.\n",
      "2019-12-17 21:48:33.831904: I tensorflow/core/platform/profile_utils/cpu_utils.cc:94] CPU Frequency: 2200000000 Hz\n",
      "2019-12-17 21:48:33.832385: I tensorflow/compiler/xla/service/service.cc:168] XLA service 0x56544440cd00 executing computations on platform Host. Devices:\n",
      "2019-12-17 21:48:33.832423: I tensorflow/compiler/xla/service/service.cc:175]   StreamExecutor device (0): Host, Default Version\n",
      "2019-12-17 21:48:33.832940: I tensorflow/core/common_runtime/process_util.cc:115] Creating new thread pool with default inter op setting: 2. Tune using inter_op_parallelism_threads for best performance.\n",
      "INFO:tensorflow:Running local_init_op.\n",
      "INFO:tensorflow:Done running local_init_op.\n",
      "INFO:tensorflow:Saving checkpoints for 0 into ./taxi_trained/model.ckpt.\n",
      "INFO:tensorflow:loss = 106.78464, step = 0\n",
      "INFO:tensorflow:global_step/sec: 111.062\n",
      "INFO:tensorflow:loss = 3.4875064, step = 100 (0.900 sec)\n",
      "INFO:tensorflow:global_step/sec: 209.705\n",
      "INFO:tensorflow:loss = 55.23517, step = 200 (0.477 sec)\n",
      "INFO:tensorflow:Saving checkpoints for 300 into ./taxi_trained/model.ckpt.\n",
      "INFO:tensorflow:Calling model_fn.\n",
      "INFO:tensorflow:Done calling model_fn.\n",
      "INFO:tensorflow:Starting evaluation at 2019-12-17T21:48:38Z\n",
      "INFO:tensorflow:Graph was finalized.\n",
      "INFO:tensorflow:Restoring parameters from ./taxi_trained/model.ckpt-300\n",
      "INFO:tensorflow:Running local_init_op.\n",
      "INFO:tensorflow:Done running local_init_op.\n",
      "INFO:tensorflow:Evaluation [5/50]\n",
      "INFO:tensorflow:Evaluation [10/50]\n",
      "INFO:tensorflow:Evaluation [15/50]\n",
      "INFO:tensorflow:Evaluation [20/50]\n",
      "INFO:tensorflow:Evaluation [25/50]\n",
      "INFO:tensorflow:Evaluation [30/50]\n",
      "INFO:tensorflow:Evaluation [35/50]\n",
      "INFO:tensorflow:Evaluation [40/50]\n",
      "INFO:tensorflow:Evaluation [45/50]\n",
      "INFO:tensorflow:Evaluation [50/50]\n",
      "INFO:tensorflow:Finished evaluation at 2019-12-17-21:48:39\n",
      "INFO:tensorflow:Saving dict for global step 300: average_loss = 7.591091, global_step = 300, label/mean = 5.318125, loss = 7.591091, prediction/mean = 2.9586215\n",
      "INFO:tensorflow:Saving 'checkpoint_path' summary for global step 300: ./taxi_trained/model.ckpt-300\n",
      "INFO:tensorflow:Calling model_fn.\n",
      "INFO:tensorflow:Done calling model_fn.\n",
      "WARNING:tensorflow:From /usr/local/lib/python3.5/dist-packages/tensorflow_core/python/saved_model/signature_def_utils_impl.py:201: build_tensor_info (from tensorflow.python.saved_model.utils_impl) is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "This function will only be available through the v1 compatibility library as tf.compat.v1.saved_model.utils.build_tensor_info or tf.compat.v1.saved_model.build_tensor_info.\n",
      "INFO:tensorflow:Signatures INCLUDED in export for Classify: None\n",
      "INFO:tensorflow:Signatures INCLUDED in export for Eval: None\n",
      "INFO:tensorflow:Signatures INCLUDED in export for Train: None\n",
      "INFO:tensorflow:Signatures INCLUDED in export for Regress: None\n",
      "INFO:tensorflow:Signatures INCLUDED in export for Predict: ['predict']\n",
      "INFO:tensorflow:Signatures EXCLUDED from export because they cannot be be served via TensorFlow Serving APIs:\n",
      "INFO:tensorflow:'regression' : Regression input must be a single string Tensor; got {'passengers': <tf.Tensor 'passengers:0' shape=(None,) dtype=float32>, 'pickuplon': <tf.Tensor 'pickuplon:0' shape=(None,) dtype=float32>, 'latdiff': <tf.Tensor 'sub_1:0' shape=(None,) dtype=float32>, 'hourofday': <tf.Tensor 'hourofday:0' shape=(None,) dtype=int64>, 'pickuplat': <tf.Tensor 'pickuplat:0' shape=(None,) dtype=float32>, 'dropofflon': <tf.Tensor 'dropofflon:0' shape=(None,) dtype=float32>, 'euclidean': <tf.Tensor 'Sqrt:0' shape=(None,) dtype=float32>, 'dayofweek': <tf.Tensor 'dayofweek:0' shape=(None,) dtype=int64>, 'dropofflat': <tf.Tensor 'dropofflat:0' shape=(None,) dtype=float32>, 'londiff': <tf.Tensor 'sub:0' shape=(None,) dtype=float32>}\n",
      "INFO:tensorflow:'serving_default' : Regression input must be a single string Tensor; got {'passengers': <tf.Tensor 'passengers:0' shape=(None,) dtype=float32>, 'pickuplon': <tf.Tensor 'pickuplon:0' shape=(None,) dtype=float32>, 'latdiff': <tf.Tensor 'sub_1:0' shape=(None,) dtype=float32>, 'hourofday': <tf.Tensor 'hourofday:0' shape=(None,) dtype=int64>, 'pickuplat': <tf.Tensor 'pickuplat:0' shape=(None,) dtype=float32>, 'dropofflon': <tf.Tensor 'dropofflon:0' shape=(None,) dtype=float32>, 'euclidean': <tf.Tensor 'Sqrt:0' shape=(None,) dtype=float32>, 'dayofweek': <tf.Tensor 'dayofweek:0' shape=(None,) dtype=int64>, 'dropofflat': <tf.Tensor 'dropofflat:0' shape=(None,) dtype=float32>, 'londiff': <tf.Tensor 'sub:0' shape=(None,) dtype=float32>}\n",
      "WARNING:tensorflow:Export includes no default signature!\n",
      "INFO:tensorflow:Restoring parameters from ./taxi_trained/model.ckpt-300\n",
      "INFO:tensorflow:Assets added to graph.\n",
      "INFO:tensorflow:No assets to write.\n",
      "INFO:tensorflow:SavedModel written to: ./taxi_trained/export/exporter/temp-b'1576619319'/saved_model.pb\n",
      "INFO:tensorflow:Loss for final step: 101.04095.\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "# Train our taxifare model locally\n",
    "rm -r ./taxi_trained\n",
    "export PYTHONPATH=${PYTHONPATH}:$PWD\n",
    "python3 -m tft_trainer.task \\\n",
    "    --train_data_path=\"gs://${BUCKET}/taxifare/preproc_tft/train*\" \\\n",
    "    --eval_data_path=\"gs://${BUCKET}/taxifare/preproc_tft/eval*\"  \\\n",
    "    --output_dir=./taxi_trained \\"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1600944292\n"
     ]
    }
   ],
   "source": [
    "# `ls` command show the full list or content of your directory\n",
    "!ls $PWD/taxi_trained/export/exporter"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's create fake data in JSON format and use it to serve a prediction with gcloud ai-platform local predict"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting /tmp/test.json\n"
     ]
    }
   ],
   "source": [
    "%%writefile /tmp/test.json\n",
    "{\"dayofweek\":0, \"hourofday\":17, \"pickuplon\": -73.885262, \"pickuplat\": 40.773008, \"dropofflon\": -73.987232, \"dropofflat\": 40.732403, \"passengers\": 2.0}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "sudo find \"/usr/lib/google-cloud-sdk/lib/googlecloudsdk/command_lib/ml_engine\" -name '*.pyc' -delete"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PREDICTIONS\n",
      "[8.322036743164062]\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "If the signature defined in the model is not serving_default then you must specify it via --signature-name flag, otherwise the command may fail.\n",
      "WARNING: WARNING:tensorflow:From /usr/local/lib/python2.7/dist-packages/tensorflow_core/python/compat/v2_compat.py:65: disable_resource_variables (from tensorflow.python.ops.variable_scope) is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "non-resource variables are not supported in the long term\n",
      "2019-12-17 21:50:07.411300: I tensorflow/core/platform/cpu_feature_guard.cc:145] This TensorFlow binary is optimized with Intel(R) MKL-DNN to use the following CPU instructions in performance critical operations:  AVX2 FMA\n",
      "To enable them in non-MKL-DNN operations, rebuild TensorFlow with the appropriate compiler flags.\n",
      "2019-12-17 21:50:07.419152: I tensorflow/core/platform/profile_utils/cpu_utils.cc:94] CPU Frequency: 2200000000 Hz\n",
      "2019-12-17 21:50:07.419442: I tensorflow/compiler/xla/service/service.cc:168] XLA service 0x560ff33e0950 executing computations on platform Host. Devices:\n",
      "2019-12-17 21:50:07.419471: I tensorflow/compiler/xla/service/service.cc:175]   StreamExecutor device (0): Host, Default Version\n",
      "2019-12-17 21:50:07.419807: I tensorflow/core/common_runtime/process_util.cc:115] Creating new thread pool with default inter op setting: 2. Tune using inter_op_parallelism_threads for best performance.\n",
      "WARNING:tensorflow:From /usr/lib/google-cloud-sdk/lib/third_party/ml_sdk/cloud/ml/prediction/frameworks/tf_prediction_lib.py:230: load (from tensorflow.python.saved_model.loader_impl) is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "This function will only be available through the v1 compatibility library as tf.compat.v1.saved_model.loader.load or tf.compat.v1.saved_model.load. There will be a new function for importing SavedModels in Tensorflow 2.0.\n",
      "WARNING:tensorflow:From /usr/lib/google-cloud-sdk/lib/third_party/ml_sdk/cloud/ml/prediction/frameworks/tf_prediction_lib.py:230: load (from tensorflow.python.saved_model.loader_impl) is deprecated and will be removed in a future version.\n",
      "Instructions for updating:\n",
      "This function will only be available through the v1 compatibility library as tf.compat.v1.saved_model.loader.load or tf.compat.v1.saved_model.load. There will be a new function for importing SavedModels in Tensorflow 2.0.\n",
      "\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "# Serve a prediction with gcloud ai-platform local predict\n",
    "model_dir=$(ls $PWD/taxi_trained/export/exporter/)\n",
    "gcloud ai-platform local predict \\\n",
    "    --model-dir=./taxi_trained/export/exporter/${model_dir} \\\n",
    "    --json-instances=/tmp/test.json"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Copyright 2022 Google Inc. Licensed under the Apache License, Version 2.0 (the \"License\"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
